package org.mq.mymq.nameserver.event.listener;

import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.HeartBeatDTO;
import org.mq.mymq.common.dto.ServiceRegistryRespDTO;
import org.mq.mymq.common.enums.NameServerResponseCode;
import org.mq.mymq.common.event.Listener;
import org.mq.mymq.nameserver.common.CommonCache;
import org.mq.mymq.nameserver.enums.ReplicationMsgTypeEnum;
import org.mq.mymq.nameserver.event.model.HeartBeatEvent;
import org.mq.mymq.nameserver.event.model.ReplicationMsgEvent;
import org.mq.mymq.nameserver.store.ServiceInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;

/**
 * @ClassName HeartBeatListener
 * @Description Broker端心跳事件监听器
 * @Author jiarong_ye
 * @Date 2025/6/19 13:24
 * @Version 1.0
 */
public class HeartBeatListener implements Listener<HeartBeatEvent> {
    private static final Logger log = LoggerFactory.getLogger(HeartBeatListener.class);

    @Override
    public void onReceive(HeartBeatEvent event) throws IllegalAccessException {
        // 把存在的实例保存下来
        ChannelHandlerContext ctx = event.getChannelHandlerContext();
        Object reqId = ctx.attr(AttributeKey.valueOf("reqId")).get();
        // 认证完成
        if (reqId == null) {
            ServiceRegistryRespDTO serviceRegistryRespDTO = new ServiceRegistryRespDTO();
            serviceRegistryRespDTO.setMsgId(event.getMsgId());
            TcpMsg tcpMsg = new TcpMsg(NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode(),
                    JSON.toJSONBytes(serviceRegistryRespDTO));
            ctx.writeAndFlush(tcpMsg);
            ctx.close();
            throw new IllegalAccessException("please registry first");
        }
        String[] brokerIdentifyArr = reqId.toString().split(":");
        ServiceInstance serviceInstance = new ServiceInstance();
        serviceInstance.setIp(brokerIdentifyArr[0]);
        serviceInstance.setPort(Integer.valueOf(brokerIdentifyArr[1]));
        serviceInstance.setLastHeartBeatTime(System.currentTimeMillis());
        HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
        heartBeatDTO.setMsgId(event.getMsgId());
        ctx.writeAndFlush(new TcpMsg(NameServerResponseCode.HEART_BEAT_SUCCESS.getCode(), JSON.toJSONBytes(heartBeatDTO)));
        CommonCache.getServiceInstanceManager().putIfExist(serviceInstance);
        ReplicationMsgEvent replicationMsgEvent = new ReplicationMsgEvent();
        replicationMsgEvent.setServiceInstance(serviceInstance);
        replicationMsgEvent.setMsgId(UUID.randomUUID().toString());
        replicationMsgEvent.setChannelHandlerContext(ctx);
        replicationMsgEvent.setType(ReplicationMsgTypeEnum.HEART_BEAT.getCode());
        CommonCache.getReplicationMsgQueueManager().put(replicationMsgEvent);
        log.info("接收到心跳数据：{}",JSON.toJSONString(event));
    }
}
